"""
@author: xiangping
@contact: xiangpingbu@gmail.com
@time: 2020/6/22 12:18 下午
@file: kafka_service
@Desc:
"""

import json
import logging
import signal
import sys
import threading
from threading import Event

from kafka import KafkaConsumer, TopicPartition, OffsetAndMetadata


class FlaskKafka():
    def __init__(self, interrupt_event, **kw):
        self.consumer = KafkaConsumer(**kw)
        self.handlers = {}
        self.interrupt_event = interrupt_event
        logger = logging.getLogger('flask-kafka-consumer')
        ch = logging.StreamHandler(sys.stdout)
        ch.setLevel(logging.INFO)
        formatter = logging.Formatter('[%(asctime)s] [%(name)s] [%(levelname)s] %(message)s')
        ch.setFormatter(formatter)
        logger.addHandler(ch)
        logger.setLevel(logging.INFO)
        self.logger = logger

    def _add_handler(self, topic, handler):
        if self.handlers.get(topic) is None:
            self.handlers[topic] = []
        self.handlers[topic].append(handler)

    def handle(self, topic):
        def decorator(f):
            self._add_handler(topic, f)
            return f

        return decorator

    def _run_handlers(self, msg):
        try:
            handlers = self.handlers[msg.topic]
            for handler in handlers:
                handler(msg)
            # self.consumer.commit()
        except Exception as e:
            self.logger.critical(str(e), exc_info=1)
            self.consumer.close()
            sys.exit("Exited due to exception")

    def signal_term_handler(self, signal, frame):
        self.logger.info("closing consumer")
        self.consumer.close()
        sys.exit(0)

    def _start(self):

        for topic in self.handlers.keys():
            ps = [TopicPartition(topic, p) for p in self.consumer.partitions_for_topic(topic)]
            offset_dict = self.consumer.end_offsets(ps)
            for p in ps:
                # 获取最新的offset
                end_offset = offset_dict[p]
                logging.info("latest offset is {}".format(end_offset))
                metadata = OffsetAndMetadata(offset=end_offset - 24, metadata="")
                # 提交offset
                # self.consumer.assign([p])
                # self.consumer.seek(p, end_offset - 100)
                self.consumer.commit({p: metadata})

        self.consumer.subscribe(topics=tuple(self.handlers.keys()))
        self.logger.info("starting consumer...registered signterm")

        for msg in self.consumer:
            self.logger.debug("TOPIC: {}, PAYLOAD: {}".format(msg.topic, msg.value))
            self._run_handlers(msg)
            # stop the consumer
            if self.interrupt_event.is_set():
                self.interrupted_process()
                self.interrupt_event.clear()

    def interrupted_process(self, *args):
        self.logger.info("closing consumer")
        self.consumer.close()
        sys.exit(0)

    def _run(self):
        self.logger.info(" * The flask Kafka application is consuming")
        t = threading.Thread(target=self._start)
        t.start()

    # run the consumer application
    def run(self):
        self._run()


INTERRUPT_EVENT = Event()

bus = FlaskKafka(INTERRUPT_EVENT,
                 bootstrap_servers="192.168.40.214:9092, 192.168.40.214:9093, 192.168.40.214:9094",
                 group_id="local_kafka",
                 value_deserializer=lambda m: json.loads(m.decode('utf-8')),
                 auto_offset_reset='latest'
                 )

def listen_kill_server():
    signal.signal(signal.SIGTERM, bus.interrupted_process)
    signal.signal(signal.SIGINT, bus.interrupted_process)
    signal.signal(signal.SIGQUIT, bus.interrupted_process)
    signal.signal(signal.SIGHUP, bus.interrupted_process)


from datetime import datetime


@bus.handle('syn_data_point')
def test_topic_handler2(msg):
    try:
        timestamp = msg.timestamp / 1000
        print(timestamp)
        if timestamp % 5 == 0:
            print("receive data point msg:" + json.dumps(msg.value))
    except Exception as e:
        logging.error('error while dump data of topic syn_data_point: ' + str(e), exc_info=1)

listen_kill_server()
bus.run()
